package priv.pfz.raft.network;

import com.google.common.collect.HashBasedTable;
import com.google.common.collect.Table;
import lombok.extern.slf4j.Slf4j;
import priv.pfz.raft.network.dto.*;
import priv.pfz.raft.peer.ClusterHolder;
import priv.pfz.raft.peer.Peer;
import priv.pfz.raft.peer.RaftEventQueue;

import java.util.Optional;
import java.util.function.BiFunction;
import java.util.function.Consumer;

import static priv.pfz.raft.peer.enums.Actions.REQ_ARRIVE;
import static priv.pfz.raft.peer.enums.Actions.RESP_ARRIVE;

/**
 * @author pengfangzhou
 * @date 2022/2/3 18:21
 */
@Slf4j
public class RpcManager {
    /**
     * 延迟表
     * fromPeer, toPeer, latency
     * latency含义
     *  -1：丢包
     *  >=0：无延迟或延迟时间（ms）
     *  默认为100ms
     */
    private static final Table<String, String, Integer> latencyTable = HashBasedTable.create();

    /**
     * 网络通信事件（req到达/resp到达）的延迟队列
     */
    private static final RaftEventQueue raftEventQueue = new RaftEventQueue();

    public static void pause() {
        raftEventQueue.pause();
    }

    public static void resume() {
        raftEventQueue.resume();
    }

    public static String status() {
        return raftEventQueue.toString();
    }

    public synchronized static void modifyLatency(String from, String to, Integer latency) {
        latencyTable.put(from, to, latency);
    }

    public synchronized static int getLatency(String from, String to) {
        return Optional.ofNullable(latencyTable.get(from, to)).orElse(100);
    }

    private static <T extends BaseReq, K> void asyncCall(T req, BiFunction<Peer, T, K> reqHandler, Consumer<K> respHandler) {
        int reqLatency = getLatency(req.getSender(), req.getReceiver());
        if (reqLatency == -1) {
            log.info("{}到{}的网络已断开", req.getSender(), req.getReceiver());
            return;
        }
        raftEventQueue.add(REQ_ARRIVE, () -> {
            Peer peer = ClusterHolder.peerMap.get(req.getReceiver());
            K resp = reqHandler.apply(peer, req);
            if (resp == null) {
                log.info("{}已停止", req.getReceiver());
                return;
            }

            int respLatency = getLatency(req.getReceiver(), req.getSender());
            if (respLatency == -1) {
                log.info("{}到{}的网络已断开", req.getReceiver(), req.getSender());
                return;
            }

            raftEventQueue.add(RESP_ARRIVE, () -> {
                respHandler.accept(resp);
            }, respLatency);
        }, reqLatency);
    }

    public static RequestVoteResp requestVote(RequestVoteReq req) {
        return null;
    }

    public static void requestVote(RequestVoteReq req, Consumer<RequestVoteResp> consumer) {
        asyncCall(req, Peer::handleRequestVote, consumer);
    }

    public static AppendEntriesResp appendEntries(AppendEntriesReq req) {
        return null;
    }

    public static void appendEntries(AppendEntriesReq req, Consumer<AppendEntriesResp> consumer) {
        asyncCall(req, Peer::handleAppendEntries, consumer);
    }
}
